package cn.itcast.dmp.runner

import cn.itcast.dmp.config.AppConfigHelper
import cn.itcast.dmp.tags.MakeTagsProcessor
import cn.itcast.dmp.untils.{DateUtils, SparkSessionUtils}
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 依据ODS表数据和商圈表数据，将数据进行标签化
  */
object DailyTagsRunner {
  // ODS表名
   val DS_TABLE_NAME: String = AppConfigHelper.AD_MAIN_TABLE_NAME
  // Area的表名
   val AREA_TABLE_NAME: String = AppConfigHelper.BUSINESS_AREAS_TABLE_NAME
  //定义历史的标签数据表名
   val HISTORY_TAGS_TABLE_NAME: String = AppConfigHelper.TAGS_TABLE_NAME_PREFIX + DateUtils.getYestdayDate()
  //定义当天的标签数据表名
  val TODAY_TAGS_TABLE =  {
    AppConfigHelper.TAGS_TABLE_NAME_PREFIX + DateUtils.getTodayDate()
  }

  def main(args: Array[String]): Unit = {
    import cn.itcast.dmp.untils.KuduUtils._
//    / 第一步、构建SparkSession实例对象
    val spark: SparkSession = SparkSessionUtils.createSparkSession(this.getClass)
    // 第二、加载ods表的数据和area表的数据，并且依据GeoHash关联

 val odsDF: DataFrame =spark.readKuduTable(DS_TABLE_NAME) match {
   case Some(df) => df
   case None => println("ERROR: ODS表无数据,结束执行"); return
 }
    // 2.2、加载area表的数据
    val areaDFOption: Option[DataFrame] = spark.readKuduTable(AREA_TABLE_NAME)
    // 2.3、ods表数据与area表数据关联
    val odsWithAreaDF:DataFrame = areaDFOption match {
      case Some(areaDF) =>
        odsDF.join(
          areaDF, //关联的表名
          odsDF.col("geoHash") === areaDF.col("geoHash"),
          "left"
        )
      case None => odsDF
    }
    // 第三、获取当日用户标签数据
    val tagsDF: DataFrame = MakeTagsProcessor.processData(odsWithAreaDF)
    tagsDF.printSchema()
    tagsDF.show(10,truncate = false)
    // 第四步、获取历史标签数据, 当获取到以后与今日标签数据进行合并
    // a. 读取历史标签数据（Kudu表读取）
//    val historyTagsDFOption: Option[DataFrame] = spark.readKuduTable(HISTORY_TAGS_TABLE_NAME)
//    historyTagsDFOption match {
//      case Some(historyDF) =>
//      // 当获取衰减完成后历史标签数据，与今日标签数据合并
//        tagsDF.union(HistoryTagsProcessor.processData(historyDF))
//      case None => tagsDF
//    }
    //第五步，保存数据  每日一张标签，不能追加，所有默认是删除
    spark.createKuduTable(TODAY_TAGS_TABLE,tagsDF.schema,Seq("main_id"))
    tagsDF.saveAsKuduTable(TODAY_TAGS_TABLE)
    // 第六、应用结束，关闭资源
    spark.stop()

  }
}
